package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo6MysqlSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //使用自定义source
    val studentDS: DataStream[(Int, String, Int, String, String)] = env.addSource(new MysqlSource())

    //统计班级人数
    studentDS
      .map(stu => (stu._5, 1))
      .keyBy(_._1)
      .sum(1)
      .print()

    env.execute()

  }

}

class MysqlSource extends SourceFunction[(Int, String, Int, String, String)] {
  /**
   * 在run方法中创建数据库连接，读取数据，将数据发送到下游时
   *
   * @param ctx ：用于将数据发送到下游
   */
  override def run(ctx: SourceFunction.SourceContext[(Int, String, Int, String, String)]): Unit = {
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //2、创建数据库连接
    val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
    //3、查询数据
    val stat: PreparedStatement = con.prepareStatement("select id,name,age,gender,clazz from student")
    //3、执行查询
    val resultSet: ResultSet = stat.executeQuery()
    //4、解析数据
    while (resultSet.next()) {
      val id: Int = resultSet.getInt("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")

      //发送数据到下游
      ctx.collect((id, name, age, gender, clazz))
    }
  }

  override def cancel(): Unit = {
  }
}